package thirdpart.bus;

import io.netty.handler.codec.mqtt.MqttQoS;
import io.vertx.core.Vertx;
import io.vertx.mqtt.MqttClient;
import lombok.NonNull;
import lombok.RequiredArgsConstructor;
import lombok.extern.log4j.Log4j2;
import thirdpart.bean.CommonMsg;
import thirdpart.codec.IMsgCodec;

import java.util.concurrent.TimeUnit;

/**
 * @Author: Tzu
 * @Date: 2021/1/3 20:17
 * @Version: 1.0
 * @Description:
 *              1、启动撮合核心（连接总线）
 *              2、发消息到总线
 */

@Log4j2
@RequiredArgsConstructor
public class MqttBusSender implements IBusSender{

    @NonNull
    private String ip;

    @NonNull
    private int port;

    @NonNull
    private IMsgCodec msgCodec;

    @NonNull
    private Vertx vertx;

    /**
     * 启动撮合核心
     */
    @Override
    public void startup() {
        //连接总线
        mqttConnect();
    }

    /**
     * 连接总线
     */
    private void mqttConnect() {
        MqttClient mqttClient = MqttClient.create(vertx);

        //发出连接
        mqttClient.connect(port,ip,res -> {
           if (res.succeeded()){
               log.info("connect to mqtt bus[ip:{},port:{}] success",ip,port);
               sender = mqttClient;
           }else {
               log.error("connect to mqtt bus[ip:{},port:{}] fail",ip,port);
               //第一次连接失败重连 隐藏循环调用的逻辑 对用户友好
               mqttConnect();
           }

        });

        //断开重连
        mqttClient.closeHandler(h -> {
            try {
                //5秒重连一次
                TimeUnit.SECONDS.sleep(5);
            }catch (Exception e){
                log.error(e);
            }
            mqttConnect();
        });
    }

    ////////////////////////////////////////////

    private volatile MqttClient sender;

    /**
     * 发送消息到总线
     * @param msg
     */
    @Override
    public void publish(CommonMsg msg) {
        sender.publish(Short.toString(msg.getMsgDst()),//目标柜台地址
                msgCodec.encodeToBuffer(msg),//把消息转码
                MqttQoS.AT_LEAST_ONCE,//消息至少到达一次
                false,//不判断是否重复
                false);//不保存
    }
}
